package com.lagou.hbase.hw;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTableWrapper;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;

public class RelationDeleteProcessor extends BaseRegionObserver {

    /**
     * 删除操作后再执行后续的对应删除操作
     */
    @Override
    public void postDelete(final ObserverContext<RegionCoprocessorEnvironment> e, final Delete delete, final WALEdit edit, final Durability durability) throws IOException {
        super.postDelete(e, delete, edit, durability);

        // 因为要操作的是同一张表，所以这里直接取同一个表名
        HTableWrapper t2 = (HTableWrapper) e.getEnvironment().getTable(TableName.valueOf(BuildTable.TABLE_NAME));

        // 获取执行删除的列族队列集合
        NavigableMap<byte[], List<Cell>> familyCellMap = delete.getFamilyCellMap();

        // 通过队列集合
        Set<Map.Entry<byte[], List<Cell>>> entries = familyCellMap.entrySet();

        // 循环队列获取cells对象
        for (final Map.Entry<byte[], List<Cell>> entry : entries) {

            List<Cell> cells = entry.getValue();

            // 循环cells对象获取每个cell的rowkey以及列族
            for (final Cell cell : cells) {
                // 本cell的rowkey字节数组（主人rowkey）
                byte[] masterRowkey = CellUtil.cloneRow(cell);
                // 列名的字节数据（朋友rowkey）
                byte[] qualifier = CellUtil.cloneQualifier(cell);

                // 判断 t2 是否存在删除的rowkey的cell（由于表设计中主人rowkey就是朋友的列名，所以这里判断朋友是否也是对方好友直接用主人rowkey）
                boolean isExists = t2.exists(new Get(qualifier).addColumn(Bytes.toBytes(BuildTable.CF_NAME), masterRowkey));

                if (isExists) {
                    // 生成一个delete对象对需要删除的列族进行删除
                    Delete delQualifier = new Delete(qualifier).addColumn(Bytes.toBytes(BuildTable.CF_NAME), masterRowkey);
                    t2.delete(delQualifier);
                }

            }
        }

        t2.close();

    }
}
